package com.shuiyun.weobo.controller;


import com.shuiyun.weobo.enums.TableNameEnum;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.apache.hadoop.hbase.filter.RowFilter;

import javax.swing.*;
import java.util.ArrayList;
import java.util.List;

/**
 * @program: weibo
 * @description:
 * @author: shuiyun
 * @create: 2020-04-26 23:11
 **/
@Api(tags = "页面操作")
@RestController
@RequestMapping("/api")
public class ApiController {

    @Autowired
    private Connection connection;

    @PostMapping("/publishWeiBo")
    @ApiOperation("发布微博")
    public String publishWeiBo(String uid,String content) throws Exception{
        Table table = connection.getTable(TableName.valueOf(TableNameEnum.TABLE_CONTENT));
        String rowKey = uid+"_"+System.currentTimeMillis();
        Put put = new Put(rowKey.getBytes());
        put.addColumn("info".getBytes(),"content".getBytes(),System.currentTimeMillis(),content.getBytes());
        table.put(put);

        //微博发布完 查看自己的粉丝有哪些
        Table relations = connection.getTable(TableName.valueOf(TableNameEnum.TABLE_RELATIONS));
        // 仅查询当前用户
        Get get = new Get(uid.getBytes());
        // 只查询粉丝列族
        get.addFamily("fans".getBytes());
        Result result = relations.get(get);
        Cell[] cells = result.rawCells();
        if (cells.length < 1){
            return "no fans";
        }

        // 定义一个集合 保存所有的粉丝
        List<byte[]> fans = new ArrayList<>();
        for (Cell cell: cells){
            byte[] bytes = CellUtil.cloneQualifier(cell);
            fans.add(bytes);
        }

        // 获取微博信箱表 保存数据
        Table receive_content = connection.getTable(TableName.valueOf(TableNameEnum.TABLE_RECEIVE_CONTENT_EMAIL));
        List<Put> putContent = new ArrayList<>();
        for (byte[] fan : fans){
            Put p = new Put(fan);
            p.addColumn("info".getBytes(), Bytes.toBytes(uid),System.currentTimeMillis(),rowKey.getBytes());
            putContent.add(p);
        }
        receive_content.put(putContent);

        return "success";
    }

    @PostMapping("/addAttends")
    @ApiOperation("增加关注")
    public String addAttends(String uid,String ...attends) throws Exception{
        Table relation = connection.getTable(TableName.valueOf(TableNameEnum.TABLE_RELATIONS));
        Put put = new Put(uid.getBytes());
        // 为该用户增加关注
        List<Put> putList = new ArrayList<>();
        for (String attend : attends){
            put.addColumn("attends".getBytes(),attend.getBytes(),attend.getBytes());
        }
        putList.add(put);
        // 为被关注的人增加粉丝
        for (String attend : attends){
            Put put1 = new Put(attend.getBytes());
            put1.addColumn("fans".getBytes(),uid.getBytes(),uid.getBytes());
            putList.add(put1);
        }
        relation.put(putList);

        // 增加关注之后 需要获取到已关注用户的所有微博 发送到自己的邮箱
        Table content = connection.getTable(TableName.valueOf(TableNameEnum.TABLE_CONTENT));
        Scan scan = new Scan();
        List<byte[]> rowkeyBytes = new ArrayList<>();
        for (String attend : attends){
            RowFilter rowFilter = new RowFilter(CompareOperator.EQUAL,new SubstringComparator(attend+"_"));
            scan.setFilter(rowFilter);
            ResultScanner scanner = content.getScanner(scan);
            for (Result result : scanner){
                byte[] row = result.getRow();
                rowkeyBytes.add(row);
            }
        }

        if (rowkeyBytes.size() > 0){
            Table receive = connection.getTable(TableName.valueOf(TableNameEnum.TABLE_RECEIVE_CONTENT_EMAIL));
            List<Put> recPuts = new ArrayList<>();
            for (byte[] rowkeyByte : rowkeyBytes){
                Put put1 = new Put(uid.getBytes());
                String rowkeyByteStr = Bytes.toString(rowkeyByte);
                String attendUid = rowkeyByteStr.substring(0, rowkeyByteStr.indexOf("_"));
                Long timestamp = Long.valueOf(rowkeyByteStr.substring(rowkeyByteStr.indexOf("_") + 1));
                System.out.println(rowkeyByteStr);
                put1.addColumn("info".getBytes(),attendUid.getBytes(),timestamp+1,rowkeyByte);
                recPuts.add(put1);
            }
            receive.put(recPuts);

        }
        return "关注成功";
    }

    @PostMapping("attendCancel")
    @ApiOperation("取消关注")
    public void attendCancel(String uid,String ...cancelAttends) throws Exception{
        Table relation = connection.getTable(TableName.valueOf(TableNameEnum.TABLE_RELATIONS));
        Table receive = connection.getTable(TableName.valueOf(TableNameEnum.TABLE_RECEIVE_CONTENT_EMAIL));

        // 删除该用户的关注  和粉丝
        List<Delete> dels = new ArrayList<>();
        Delete delete = new Delete(uid.getBytes());
        Delete receiveDel = new Delete(uid.getBytes());
        for (String cancelAttend : cancelAttends){
            Delete fanDel = new Delete(cancelAttend.getBytes());
            fanDel.addColumn("fans".getBytes(),uid.getBytes());
            dels.add(fanDel);
            delete.addColumn("attends".getBytes(),cancelAttend.getBytes());
            receiveDel.addColumn("info".getBytes(),cancelAttend.getBytes());
        }
        dels.add(delete);
        relation.delete(dels);
        receive.delete(receiveDel);

        //关闭
        relation.close();
        receive.close();
    }

    @GetMapping("getContent")
    @ApiOperation("获取朋友圈")
    public List<String> getContent(String uid) throws Exception{
        List<String> resultStr = new ArrayList<>();
        Table receive = connection.getTable(TableName.valueOf(TableNameEnum.TABLE_RECEIVE_CONTENT_EMAIL));
        List<Get> rowKeyList = new ArrayList<>();
        Get get = new Get(uid.getBytes());
        get.setMaxVersions(5);
        Result result = receive.get(get);
        Cell[] cells = result.rawCells();
        for (Cell cell : cells){
            byte[] bytes = CellUtil.cloneValue(cell);
            rowKeyList.add(new Get(bytes));
        }
        // 读取内容
        Table content = connection.getTable(TableName.valueOf(TableNameEnum.TABLE_CONTENT));
        Result[] results = content.get(rowKeyList);
        for (Result result1 : results){
            byte[] value = result1.getValue("info".getBytes(), "content".getBytes());
            byte[] row = result1.getRow();
            String rowkey = Bytes.toString(row);
            String[] split = rowkey.split("_");
            resultStr.add(split[0]+":"+split[1]+":"+Bytes.toString(value));
        }
        return resultStr;
    }


}
